package com.fanghuaiming.tsec.order.mqservice;

import com.alibaba.fastjson.JSONObject;
import com.fanghuaiming.tsec.order.common.Jms;
import com.fanghuaiming.tsec.order.dao.model.TsecOrder;
import com.fanghuaiming.tsec.order.service.OrderService;
import java.math.BigDecimal;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/****
 * @description: 订单服务监听mq
 * @version:1.0.0
 * @author fanghuaiming
 * @data Created in 2020/12/1 下午5:57
 *
 */
@Slf4j
@Component
public class OrderConsumer {

    private DefaultMQPushConsumer consumer;

    /**
     * @Description: 订单消费消息（spring自动启动运行）
     *
     * @param:
     * @return:
     * @auther: fanghuaiming
     * @date: 2020/12/1 下午6:20
     * @version:1.0.0
     */
    public OrderConsumer(@Autowired Jms jms, @Autowired OrderService orderService) throws MQClientException {
        //设置消费组
        consumer = new DefaultMQPushConsumer(jms.getConsumerGroup());
        //从队列尾部开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 添加服务器地址
        consumer.setNamesrvAddr(jms.getNameServer());
        // 添加订阅topic
        consumer.subscribe(jms.getPayTopic(), "*");
        // 监听消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            MessageExt msg = msgs.get(0);
            String message = new String(msgs.get(0).getBody());
            JSONObject jsonObject = JSONObject.parseObject(message);
            BigDecimal money = jsonObject.getBigDecimal("money");
            Long userId = jsonObject.getLong("userId");
            String goods = jsonObject.getString("goods");
            String time = jsonObject.getString("time");
            //支付服务发送消息里面的唯一uuid,可以做成雪花算法进行幂等业务操作
            String key = msg.getKeys();
            log.info("========== 消费端消费消息，订单金额:{},用户ID:{},商品名称:{},分布式事务唯一ID:{},时间:{} ==========",money,userId,goods,key,time);
            try {
                orderService.createOrder(TsecOrder.builder().money(money).goods(goods).userId(userId).build());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            } catch (Exception e) {
                log.info("========== 消费失败,进行重试,重试到一定次数 那么将该条记录记录到数据库中或者放入mq补偿队列,进行人工处理 ==========");
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        consumer.start();
        log.info("========== OrderConsumer 已经启动 ==========");
    }
}
